/*
 * Copyright (C) 2015 kmy-framework-base Project
 *               Author: Administrator
 *               Date: 2015年11月4日
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.zmq.ha.client.core;

import java.io.IOException;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ShutdownListener;
import com.zmq.ha.client.declaration.Binding;
import com.zmq.ha.client.declaration.QueueDeclaration;
import com.zmq.ha.client.declaration.ResourceDeclaration;
import com.zmq.ha.client.facotry.NameThreadFacotry;
import com.zmq.ha.client.listner.MyConnectionShutdownListener;
import com.zmq.ha.client.resource.ConnectionOption;
import com.zmq.ha.client.resource.RetryableSource;


public class ConnectionWrapper extends RetryableSource implements InvocationHandler {
	
	private static final Class<?>[] CHANNEL_TYPES = {Channel.class};

	public static final ExecutorService COMSUMER_EXECUTRO = Executors.newCachedThreadPool(new NameThreadFacotry("rabbimtmq.recovery"));
	
	private Connection delegate;
	
	private ExecutorService consumerThreadPool;
	
	private ConnectionOption connectionOption;
	
	public final ConcurrentHashMap<String, ResourceDeclaration> nameMapExchangeDeclaration = new ConcurrentHashMap<String, ResourceDeclaration>();
	
	public final ConcurrentHashMap<Integer, ChannelWrapper> channelHandlers = new ConcurrentHashMap<Integer,ChannelWrapper>();
	
	public final ConcurrentHashMap<String, QueueDeclaration> nameMapQueueDeclarations = new ConcurrentHashMap<String, QueueDeclaration>();
	
	public final ConcurrentHashMap<String, ResourceDeclaration> exchangeDeclarations = new ConcurrentHashMap<String, ResourceDeclaration>();
	
	public final ConcurrentHashMap<String, ArrayList<Binding>> exchangeBindings = new ConcurrentHashMap<String, ArrayList<Binding>>();
	
	public final ConcurrentHashMap<String, ArrayList<Binding>> queueBindings = new ConcurrentHashMap<String, ArrayList<Binding>>();
	
	public Channel recoveryChannel;
	
	static final int RECOVERY_CHANNEL_NUM = 100;
	
	public static final ReentrantLock reentrantLock = new ReentrantLock();
	public static final Condition connectionRecoveryCondition = reentrantLock.newCondition();
	
	public volatile boolean isRecovery = false;
	
	public ConnectionWrapper(ConnectionOption connectionOption) {
		super();
		this.connectionOption = connectionOption;
	}

	@Override
	public Object invoke(Object proxy, final Method method, final Object[] args) throws Throwable {
		 return  deal(new Callable<Object>() {
			 
			@Override
			public Object call() throws Exception {
				
				//TODO 如果是正在恢复过程中，阻塞所有流程,等待恢复完成
				if( isRecovery ){
					ConnectionWrapper.reentrantLock.lock();
					try{
						ConnectionWrapper.connectionRecoveryCondition.await();
					}finally{
						ConnectionWrapper.reentrantLock.unlock();
					}
				}
				
				String methodName = method.getName();
				if( methodName.equals("createChannel") ){
					Channel channel = (Channel) method.invoke(delegate, args);
					ChannelWrapper channelHandler = new ChannelWrapper(channel, ConnectionWrapper.this);
					Channel proxy = (Channel) Proxy.newProxyInstance(Connection.class.getClassLoader(), CHANNEL_TYPES, channelHandler);
					channelHandler.proxy = proxy;
					channelHandlers.put(channel.getChannelNumber(), channelHandler);
					log.info("build channel [ channel:"+channel.getChannelNumber()+" ]");
					return proxy;
				}
				return method.invoke(delegate, args);
			}
		},false);
	}
	
	public void createConnection( final boolean isRecovery ) throws IOException, TimeoutException{
		
		try {
			this.consumerThreadPool = connectionOption.getConsumerPool() == null ? Executors.newCachedThreadPool(new NameThreadFacotry("rabbitmq.connection"))
					: connectionOption.getConsumerPool();
			connectionOption.setConsumerPool(this.consumerThreadPool);
			
			
			createConnection(connectionOption, isRecovery);
			ShutdownListener myShutdownListener = new MyConnectionShutdownListener(this);
			delegate.addShutdownListener(myShutdownListener);
		} catch (Exception e) {
			log.error("Connect RMQ  fail", e);
			this.connectionClose();
			throw e;
		}
	}
	
	
	public void createConnection( final ConnectionOption option , final boolean isRecovery ) throws IOException, TimeoutException{
		log.debug("building connection");
		try {
			delegate = deal(new Callable<Connection>() {
				@Override
				public Connection call() throws Exception {
					ConnectionFactory connectionFactory = option.getConnectionFactory();
					return connectionFactory.newConnection(consumerThreadPool, option.getAddresses());
				}
			}, isRecovery);
		} catch (Throwable t) {
	      if (t instanceof IOException)
	          throw (IOException) t;
	        if (t instanceof TimeoutException)
	          throw (TimeoutException)t;
	        if (t instanceof RuntimeException)
	          throw (RuntimeException) t;
	      }
		
	}
	
	public void recoveryConnection(){
		try {
			//重新建立connection
			this.createConnection(true);
			
			this.recoveryExchangeAndBinding();
			this.recoveryQueueAndBinding();
			
			for (ChannelWrapper channelHandler : channelHandlers.values()){
				channelHandler.recoverChannel(true);
			}
			
		} catch (Exception e) {
			log.error( "recovering Connection fail ...", e );
		}finally{
			this.isRecovery = false;
			ConnectionWrapper.reentrantLock.lock();
			try{
				ConnectionWrapper.connectionRecoveryCondition.signalAll();
			}finally{
				ConnectionWrapper.reentrantLock.unlock();
			}
		}
	}
	
	
	public void recoveryExchangeAndBinding(){
		if( nameMapExchangeDeclaration.size() > 0 && exchangeBindings.size() > 0 ){
			for( Map.Entry<String, ResourceDeclaration> entry : nameMapExchangeDeclaration.entrySet() ){
				try {
					recoverExchange(entry.getKey(), entry.getValue());
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
		try {
			recoverExchangeBindings(exchangeBindings.values().iterator());
		} catch (Exception e) {
			log.error( "Exception occur when recovering ExchangeBinding...", e );
		}
	}
	
	
	public void recoveryQueueAndBinding(){
		if( nameMapQueueDeclarations.size() > 0 && queueBindings.size() > 0  ){
			for( Map.Entry<String, QueueDeclaration> entry : nameMapQueueDeclarations.entrySet()  ){
				try {
					recoverQueue(entry.getKey(), entry.getValue());
				} catch (Exception e) {
					log.error(" Exception occur when recovering QueueDeclarations...",e);
				}
				
			}
		}
		
		try {
			recoverQueueBindings(queueBindings.values().iterator() );
		} catch (Exception e) {
			log.error(" Exception occur when recovering QueueBindings...",e);
		}
	}
	
	
	 /** Updates the queue name referenced by queue bindings. */
	  void updateQueueBindingReferences(String oldQueueName, String newQueueName) {
	    ArrayList<Binding> bindings = queueBindings.remove(oldQueueName);
	    if (bindings != null) {
	      for (Binding binding : bindings)
	        binding.destination = newQueueName;
	      queueBindings.put(newQueueName, bindings);
	    }
	  }
	

	/**
	 * 获取需要恢复的Channel,如果当前没有需要恢复的Channel，则新建一个。
	 */
	@Override
	public Channel getRecoveryChannel() throws IOException {
		if( recoveryChannel == null && !recoveryChannel.isOpen() )
			return delegate.createChannel( RECOVERY_CHANNEL_NUM ); //该处创立的channel之后会关闭，只为了临时绑定queue，exchange.
		return recoveryChannel;
	}

	@Override
	public boolean throwOnRecoveryFailure() {
		// TODO Auto-generated method stub
		return false;
	}
	
	 Channel createChannel(int channelNumber) throws IOException {
		    return delegate.createChannel(channelNumber);
	 }
	 
	 public void connectionShutdown() {
	    for (ChannelWrapper channelWrapper : channelHandlers.values())
	    	channelWrapper.channelShutdown();
	  }
	 
	 /**
	  * 
	  * @Title: connectionClose
	  * @Description: 关闭RMQ Connection
	  * @param      
	  * @return void     
	  * @throws
	  */
	 public void connectionClose(){
		 this.consumerThreadPool.shutdown();
	 }

}
